package com.bfl.mq.service.impl;

import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendCallback;
import com.alibaba.rocketmq.common.message.Message;
import com.bfl.mq.client.entity.MQEntity;
import com.bfl.mq.service.IProducer;
import com.bfl.mq.utils.SerializableUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;

import java.util.UUID;

public class RocketMqProducerImpl implements IProducer, InitializingBean {

    private static Logger logger = LoggerFactory.getLogger(RocketMqProducerImpl.class);

    private String namesrvAddr;

    private String producerGroup;

    private Boolean retryAnotherBrokerWhenNotStoreOK;

    private DefaultMQProducer producer;

    /**
     * spring 容器初始化所有属性后调用此方法
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        producer = new DefaultMQProducer();
        producer.setProducerGroup(this.producerGroup);
        producer.setNamesrvAddr(this.namesrvAddr);
        producer.setRetryAnotherBrokerWhenNotStoreOK(this.retryAnotherBrokerWhenNotStoreOK);
        // producer.setVipChannelEnabled(false);
        /*
         * Producer对象在使用之前必须要调用start初始化，初始化一次即可<br>
         * 注意：切记不可以在每次发送消息时，都调用start方法
         */
        producer.start();
        logger.info("[{}:{}] start successd!", producerGroup, namesrvAddr);
    }

    /**
     * 销毁
     */
    public void destroy() throws Exception {
        if (producer != null) {
            logger.info("producer: [{}:{}] end ", producerGroup, namesrvAddr);
            producer.shutdown();
        }

    }

    @Override
    public void send(String topic, MQEntity entity) {
        String keys = UUID.randomUUID().toString();
        entity.setMqKey(keys);
        String tags = entity.getClass().getName();
        logger.info("业务:{},tags:{},keys:{},entity:{}",topic, tags, keys, entity);
        Message msg = new Message(topic, tags, keys,
                SerializableUtil.toByte(entity));
        try {
            producer.send(msg);
        } catch (Exception e) {
            logger.error(keys.concat(":发送消息失败"), e);
            throw new RuntimeException("发送消息失败",e);
        }
    }

    @Override
    public void send(String topic, MQEntity entity, SendCallback sendCallback) {
        String keys = UUID.randomUUID().toString();
        entity.setMqKey(keys);
        String tags = entity.getClass().getName();
        logger.info("业务:{},tags:{},keys:{},entity:{}",topic, tags, keys, entity);
        Message msg = new Message(topic, tags, keys,
                SerializableUtil.toByte(entity));
        try {
            producer.send(msg, sendCallback);
        } catch (Exception e) {
            logger.error(keys.concat(":发送消息失败"), e);
            throw new RuntimeException("发送消息失败",e);
        }
    }

    @Override
    public void sendOneway(String topic, MQEntity entity) {
        String keys = UUID.randomUUID().toString();
        entity.setMqKey(keys);
        String tags = entity.getClass().getName();
        logger.info("业务:{},tags:{},keys:{},entity:{}",topic, tags, keys, entity);
        Message msg = new Message(topic, tags, keys,
                SerializableUtil.toByte(entity));
        try {
            producer.sendOneway(msg);
        } catch (Exception e) {
            logger.error(keys.concat(":发送消息失败"), e);
            throw new RuntimeException("发送消息失败",e);
        }
    }

    public void setNamesrvAddr(String namesrvAddr) {
        this.namesrvAddr = namesrvAddr;
    }

    public void setProducerGroup(String producerGroup) {
        this.producerGroup = producerGroup;
    }

    public void setRetryAnotherBrokerWhenNotStoreOK(Boolean retryAnotherBrokerWhenNotStoreOK) {
        this.retryAnotherBrokerWhenNotStoreOK = retryAnotherBrokerWhenNotStoreOK;
    }

    public void setProducer(DefaultMQProducer producer) {
        this.producer = producer;
    }
}
